/**
 * QUANSHI.com Inc.
 * Copyright (c) 2016-2017 All Rights Reserved.
 */
package com.quanshi.scheduler.job.biz.route.strategy;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

import com.quanshi.scheduler.core.base.Ret;
import com.quanshi.scheduler.core.biz.model.TriggerParam;
import com.quanshi.scheduler.job.biz.route.ExecutorRouter;
import com.quanshi.scheduler.job.dal.entity.TriggerLog;

/**
 * 最不经常使用
 * 
 * @author yanxiang.huang 2017-07-10 13:41:09
 */
public class ExecutorRouteLFU extends ExecutorRouter
{
    private static ConcurrentHashMap<Long, HashMap<String, Integer>> jobLfuMap = new ConcurrentHashMap<Long, HashMap<String, Integer>>();
    private static long CACHE_VALID_TIME = 0;

    @Override
    public String route( long jobId, ArrayList<String> addressList )
    {
        // cache clear
        if (System.currentTimeMillis() > CACHE_VALID_TIME) {
            jobLfuMap.clear();
            CACHE_VALID_TIME = System.currentTimeMillis() + 1000 * 60 * 60 * 24;
        }

        // lfu item init
        // Key排序可以用TreeMap+构造入参Compare; Value排序暂时只能通过ArrayList；
        HashMap<String, Integer> lfuItemMap = jobLfuMap.get(jobId);     
        if (lfuItemMap == null) {
            lfuItemMap = new HashMap<String, Integer>();
            jobLfuMap.put(jobId, lfuItemMap);
        }
        for (String address: addressList) {
            if (!lfuItemMap.containsKey(address) || lfuItemMap.get(address) >1000000 ) {
                // 初始化时主动Random一次，缓解首次压力
                lfuItemMap.put(address, new Random().nextInt(addressList.size()));  
            }
        }

        // load least used count address
        List<Map.Entry<String, Integer>> lfuItemList = new ArrayList<Map.Entry<String, Integer>>(lfuItemMap.entrySet());
        Collections.sort(lfuItemList, new Comparator<Map.Entry<String, Integer>>() {
            @Override
            public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
                return o1.getValue().compareTo(o2.getValue());
            }
        });

        Map.Entry<String, Integer> addressItem = lfuItemList.get(0);
        addressItem.setValue(addressItem.getValue() + 1);

        return addressItem.getKey();
    }

    @Override
    public Ret<String> routeRun( TriggerParam triggerParam, ArrayList<String> addressList, TriggerLog jobLog )
    {
        // address
        String address = route(triggerParam.getJobId(), addressList);
        jobLog.setExecutorAddress(address);

        // run executor
        Ret<String> runResult = runExecutor(triggerParam, address);
        runResult.setMsg("<br>----------------------<br>" + runResult.getMsg());

        return runResult;
    }

}
